package com.surfilter.massdata.spark.task.daystat2;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.storage.StorageLevel;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.DnsCount;
import com.surfilter.massdata.spark.bean.DnsDomain;
import com.surfilter.massdata.spark.bean.IndustryDomain;
import com.surfilter.massdata.spark.bean.SA7601;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.ConfigUtil;
import com.surfilter.massdata.spark.util.DateUtil;


public class DomainAnalyzTask_SA7601_ClassdayNew implements ISparkTask{
	private static final Log log = LogFactory.getLog(DomainAnalyzTask_SA7601_ClassdayNew.class);
	private static final long serialVersionUID = 1L;
	private static String outName;

	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames, Map<String, String> commandMap) {
		long start=System.currentTimeMillis();
		long temp=start;
		long temp_end=start;
		double exectime=0.0;
		
		System.out.println("==============DomainAnalyzTask_SA7601 begin====================");
		System.out.println("............网站分类....开始............");
		
		String dayStr = commandMap.get("-d");
        Date date = DateUtil.getExecDate(DateUtil.getYesterday(),dayStr);
		
		List<OutQueueEntity> outList = new ArrayList<OutQueueEntity>();
		try{
			String p_linces_key = new String(ConfigUtil.get("LICENSE_KEY").getBytes("ISO-8859-1"),"UTF-8");
			log.info("LICENSE_KEY="+p_linces_key);
			DataFrame dataSource = dataFrames.get("dws_hour_sa7601");
			
			String classsql="select domain,sum(traffic)as count  from dws_hour_sa7601_temp where  BUSS_TYPE='WEBSITE_RANK_BUSS' and sta_range=0  GROUP BY  domain";
			DataFrame df_sa7601=dataSource.sqlContext().sql(classsql);
			JavaRDD<DnsCount> dnsCountRdd = df_sa7601.toJavaRDD().mapPartitions(new FlatMapFunction<Iterator<Row>,DnsCount>(){

				@Override
				public Iterable<DnsCount> call(Iterator<Row> it) throws Exception {
					List<DnsCount> list = new ArrayList<DnsCount>(); 
					while(it.hasNext())
					{
						Row row = it.next();
						String top_domain =row.getAs("domain").toString();
						long count=Long.parseLong(row.getAs("count").toString());
						if(StringUtils.isNoneBlank(top_domain))
						{
							DnsCount sa = new DnsCount();
							sa.setTop_domain(top_domain);
							sa.setCount(count);
							list.add(sa);
						}
					}
					return list;
				}
				
			});
	        
	    	DataFrame df_dc = dataSource.sqlContext().createDataFrame(dnsCountRdd, DnsCount.class);
	    	df_dc.registerTempTable("df_dc_temp");
	        df_dc.persist(StorageLevel.MEMORY_AND_DISK_SER());
			String joinsql = "select a.top_domain,a.count, "
					+ "(CASE WHEN ISNULL(b.license_key) = false THEN B.license_key ELSE  'N/A' END) AS license_key,"
					+ "(CASE WHEN ISNULL(b.site_name) = false THEN b.site_name ELSE  'N/A' END) AS site_name, "
					+ "(CASE WHEN ISNULL(b.PARENT_INDUSTRY_TYPE) = false THEN b.PARENT_INDUSTRY_TYPE ELSE  '221' END) AS parent_industry_type "
					+ "from df_dc_temp a left join br2005_temp b on a.top_domain = b.top_domain order by b.parent_industry_type,a.count desc "; 
			DataFrame oper_tmp = dataSource.sqlContext().sql(joinsql);
			oper_tmp.registerTempTable("df_type_temp");
			oper_tmp.persist(StorageLevel.MEMORY_AND_DISK_SER());
			//List<Row> top_domain_100 = oper_tmp.toJavaRDD().take(100);
	        //String sql_br2005_industry="select distinct parent_industry_type from br2005_temp";
			//从BR2005表中获取parent_industry_type有可能丢失221和空的行业类型,改为从关联结果表中获取
			/*依据parent_industry_type，对数据进行拆分，例子：(na018,na004), 将被拆成两条*/
			log.info("parent_industry_type=(na018,na004)");
			JavaRDD<DnsDomain> dnsDomainRdd=oper_tmp.toJavaRDD().mapPartitions(new FlatMapFunction<Iterator<Row>,DnsDomain>(){

				@Override
				public Iterable<DnsDomain> call(Iterator<Row> it) throws Exception {
					List<DnsDomain> list = new ArrayList<DnsDomain>(); 
					while(it.hasNext())
					{
						Row row = it.next();
						String top_domain =row.getAs("top_domain").toString();
						long count=Long.parseLong(row.getAs("count").toString());
						String type_num = row.getAs("parent_industry_type");
			    		if(type_num == null || type_num.equals("") ){
			    			type_num="221";
			    		}
			    		String [] line =StringUtils.split(type_num, ",");
			    		for(String t:line){
			    			
			    			if(StringUtils.isNoneBlank(top_domain))
							{
								DnsDomain sa = new DnsDomain();
								sa.setTop_domain(top_domain);
								sa.setCount(count);
								sa.setLicense_key(row.getAs("license_key").toString());
								sa.setParent_industry_type(t);
								sa.setSite_name(row.getAs("site_name").toString());
								list.add(sa);
							}
			    		}
					}
					return list;
				}
				
			});
			DataFrame df_dnsdomain = dataSource.sqlContext().createDataFrame(dnsDomainRdd, DnsDomain.class);
			df_dnsdomain.registerTempTable("df_dns_domain_temp");
			df_dnsdomain.persist(StorageLevel.MEMORY_AND_DISK_SER());
	        
	        String sql_br2005_industry="select distinct parent_industry_type from df_dns_domain_temp";
	    	DataFrame industry_type_df = dataSource.sqlContext().sql(sql_br2005_industry);
	    	List<SA7601> list_SA7601 = new ArrayList<SA7601>();	
	    	List<Row> type_list=industry_type_df.toJavaRDD().collect();
	    	for(int i=0;i<type_list.size();i++){
	    		String type_num = type_list.get(i).getAs("parent_industry_type");
	    		if(type_num == null || type_num.equals("") ){
	    			type_num="221";
	    		}
	    		final String type=type_num;
	    		List<DnsDomain> industryDomainList=dnsDomainRdd.filter(new Function<DnsDomain, Boolean>(){
					private static final long serialVersionUID = 1L;
					@Override
					public Boolean call(DnsDomain v1) throws Exception {
						String industry_type=v1.getParent_industry_type();
						if(StringUtils.equals(type, industry_type)){
							return true;
						}
						return false;
					}
				}).take(100);
	    		
	    		//type_list
	    		if(industryDomainList.size()>0){
	    			for(int j=0;j<industryDomainList.size();j++){
	    				SA7601 sa = new SA7601();
	    				DnsDomain trow = industryDomainList.get(j);
	    				sa.setYear(DateUtil.getCurrentYear(date));
		        		sa.setHalf_year(DateUtil.getHalfYear(date));
		        		sa.setQuarter(DateUtil.getQuarter(date));
		        		sa.setMonth(DateUtil.getCurrentMonth(date));
		        		sa.setWeek(DateUtil.getCurrentWeek(date));
		        		sa.setDay(DateUtil.getCurrentDay(date));
		        		sa.setBuss_type("WEBSITE_RANK_BUSS");
		        		sa.setBuss_value(trow.getParent_industry_type());
		        		sa.setRank(j + 1);
		        		sa.setDomain(trow.getTop_domain());
		        	    sa.setWebsite_name(trow.getSite_name());
		        	    sa.setIcp_num(trow.getLicense_key());
		        	    sa.setTraffic(String.valueOf(trow.getCount()));
		        		sa.setSta_range(0);
		        		list_SA7601.add(sa);
	    			}
	    		}
	    	}
	    	df_dc.sqlContext().dropTempTable("df_dc_temp");
		    df_dc.unpersist();
		    df_dnsdomain.sqlContext().dropTempTable("df_dns_domain_temp");
		    df_dnsdomain.unpersist();
		    
	    	System.out.println("网站分类 exectime: "+exectime+" min........................");
	    	JavaSparkContext ctx = new JavaSparkContext(dataSource.sqlContext().sparkContext());
	    	JavaRDD<SA7601> rdd = ctx.parallelize(list_SA7601);
	    	DataFrame df = dataSource.sqlContext().createDataFrame(rdd, SA7601.class);
	    	outList.add(new OutQueueEntity(outName, df));
			
			CommonUtils.deleteTaskTableData("SA7601", date, "WEBSITE_RANK_BUSS", 0, "day");
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		
		long end=System.currentTimeMillis();
		double min=(end-start)*1.0/(1000*60);
		System.out.println("DomainAnalyzTask_SA7601_New分类网站访问量排名:exectime: "+min+" min............");
		return outList;
	}
}
